#pragma once
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include <iostream>
#include <functional>
#include "logger.hpp"
//这里用到了一个AMQP协议，一个AMQP-CPP库一个RabbitMQ服务，一个libev组件
//其中AMQP协议是用来进行TCP客户端之间的通信协议
//AMQP-CPP库用来使用AMQP协议进行异步I/O（AMQP-CPP库是完全异步的且不需要使用线程）
//RabbitMQ服务用来对协议进行序列化和反序列化
//libev用来对使用AMQP-CPP库的I/O进行监控，并提供了异步I/O的结果处理回调函数框架
namespace zzy_mims
{

const std::string grouting_key = "routing_key";
const std::string gconsume_tag = "consume_tag";
class RMQClient
{
public:
    using ptr = std::shared_ptr<RMQClient>;
    using MessageCB_t = std::function<void(const std::string)>;
    //"amqp://root:123456@127.0.0.1:5672/"
    RMQClient(const std::string& user,const std::string& password,const std::string& host)
    :_handler(_loop),_connection(&_handler,AMQP::Address("amqp://"+user+":"+password+"@"+host+"/")),
    _channel(&_connection),_loop_thread(std::thread([this](){ev_run(_loop,0);}))
    {};
    ~RMQClient()
    {
        ev_async async_watcher;
        ev_async_init(&async_watcher,watcher_callback);
        ev_async_start(_loop,&async_watcher);
        ev_async_send(_loop, &async_watcher);
        _loop_thread.join();
        _loop = nullptr;
    }
    void declareComponents(const std::string& exchange_name,const std::string& queue_name,
    const std::string& routing_key=grouting_key)
    {
        //5.声明交换机
        _channel.declareExchange(exchange_name,AMQP::ExchangeType::direct)
        .onError([](const char* message)
        {
            LOG_ERROR("声明交换机失败:{}\n",message);
            exit(0);
        })
        .onSuccess([]()
        {
            LOG_DEBUG("声明交换机成功\n");
        });
        //6.声明交换队列
        _channel.declareQueue(queue_name)
        .onError([](const char* message)
        {
            LOG_ERROR("声明交换队列失败:{}\n",message);
            exit(0);
        })
        .onSuccess([]()
        {
            LOG_DEBUG("声明交换队列成功\n");
        });
        //7.管道的交换机和队列进行绑定
        _channel.bindQueue(exchange_name,queue_name,routing_key)
        .onError([](const char* message)
        {
            LOG_ERROR("管道的交换机和队列绑定失败:{}\n",message);
            exit(0);
        })
        .onSuccess([]()
        {
            LOG_DEBUG("管道的交换机和队列绑定\n");
        });
    }
    bool publish(const std::string &exchange_name, const std::string &msg, 
            const std::string &routing_key = grouting_key)
        {
            LOG_DEBUG("向交换机{}-{}发送消息\n",exchange_name,routing_key);
            if(_channel.publish(exchange_name,routing_key,msg)==false)
            LOG_ERROR("向交换机{}-{}发送消息失败\n",exchange_name,routing_key);
            return true;
        }
    void consume(const std::string& queue_name,const MessageCB_t& mcb)
    {
        LOG_DEBUG("开始订阅{}队列消息",queue_name);
        _channel.consume(queue_name,gconsume_tag)
        .onReceived([this,mcb](const AMQP::Message& message,uint64_t deliveryTag, bool redelivered)
        {
            mcb(std::string(message.body(),message.bodySize()));
            _channel.ack(deliveryTag);
        })
        .onError([queue_name](const char* message)
        {
            LOG_ERROR("订阅{}队列消息失败:{}\n",queue_name,message);
            exit(0);
        });
    }
private:
    static void watcher_callback(struct ev_loop *loop, ev_async *watcher, int32_t revents)
    {
        ev_break(loop,EVBREAK_ALL);
    }
    struct ev_loop* _loop= EV_DEFAULT;
    AMQP::LibEvHandler _handler;
    AMQP::TcpConnection _connection;
    AMQP::TcpChannel _channel;
    std::thread _loop_thread;
};
}